package cn.bearspark.tag.provider.dao.repository;

import cn.bearspark.common.dto.Page;
import cn.bearspark.tag.provider.dao.po.TagIdContentPO;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Repository;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 标签内容数据的数据仓库对象
 *
 * @author f
 */
@Repository
public class TagContentRepository {

    private static final Logger LOGGER = LoggerFactory.getLogger(TagContentRepository.class);

    /**
     * 标签的索引名称
     */
    private static final String INDEX = "tag";

    /**
     * 标签内容的字段名称，同时表示 在 Java 对象中的字段 和 在 es 中的字段
     */
    private static final String CONTENT_FIELD_NAME = "content";

    /**
     * 查询时用于排序的字段名称，表示在 es 中的字段
     */
    private static final String TOP_SCORE_FIELD_NAME = "topScore";

    /**
     * 包含标签内容的包含组
     */
    private static final String[] CONTENT_INCLUDES = {CONTENT_FIELD_NAME};

    /**
     * 空的 String 数组
     */
    private static final String[] NULL_ARRAY = {};

    private final RestHighLevelClient restHighLevelClient;

    public TagContentRepository(RestHighLevelClient restHighLevelClient) {
        this.restHighLevelClient = restHighLevelClient;
    }

    // 将 long 类型的集合转化为 String 类型的数组
    private static String[] toStringList(List<Long> idList) {
        return idList.stream().map(String::valueOf).toArray(String[]::new);
    }

    /**
     * 从 es 中获取指定标签的内容
     *
     * @param tagIdList 指定标签的 id 集合
     * @return 指定标签的内容映射，key：标签的 id，value：标签的内容
     */
    public Map<Long, String> mapContent(List<Long> tagIdList) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
                .fetchSource(CONTENT_INCLUDES, NULL_ARRAY)
                .query(QueryBuilders.idsQuery().addIds(toStringList(tagIdList)));

        SearchRequest searchRequest = new SearchRequest()
                .indices(INDEX)
                .source(sourceBuilder);

        try {
            SearchHit[] hits = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT)
                    .getHits().getHits();
            Map<Long, String> map = new HashMap<>(tagIdList.size());
            for (SearchHit hit : hits) {
                map.put(Long.valueOf(hit.getId()), hit.getSourceAsMap().get(CONTENT_FIELD_NAME).toString());
            }
            return map;
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
            return null;
        }
    }

    /**
     * 条件分页查询，查询指定数量的数据
     *
     * @param keyword  关键字
     * @param pageNum  页数
     * @param pageSize 元素数
     * @return 条件分页查询的数据
     */
    public Page<TagIdContentPO> pageQuery(String keyword, int pageNum, int pageSize) {
        SearchSourceBuilder sourceBuilder;
        if (StringUtils.hasText(keyword)) {
            sourceBuilder = new SearchSourceBuilder()
                    .fetchSource(CONTENT_INCLUDES, NULL_ARRAY)
                    .query(QueryBuilders.matchQuery(CONTENT_FIELD_NAME, keyword))
                    .highlighter(new HighlightBuilder().field(CONTENT_FIELD_NAME))
                    .sort(TOP_SCORE_FIELD_NAME, SortOrder.DESC)
                    .from((pageNum - 1) * pageSize)
                    .size(pageSize);
        } else {
            // 如果 keyword 为空，则查询所有的标签
            sourceBuilder = new SearchSourceBuilder()
                    .fetchSource(CONTENT_INCLUDES, NULL_ARRAY)
                    .query(QueryBuilders.matchAllQuery())
                    .sort(TOP_SCORE_FIELD_NAME, SortOrder.DESC)
                    .from((pageNum - 1) * pageSize)
                    .size(pageSize);
        }

        SearchRequest searchRequest = new SearchRequest()
                .indices(INDEX)
                .source(sourceBuilder);

        try {
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = searchResponse.getHits();
            List<TagIdContentPO> tagIdContentPOList = new ArrayList<>(hits.getHits().length);
            for (SearchHit hit : hits) {
                tagIdContentPOList.add(fromHighlightHit(hit));
            }

            Page<TagIdContentPO> tagIdContentPOPage = new Page<>();
            tagIdContentPOPage.setTotal(hits.getTotalHits().value);
            tagIdContentPOPage.setRecords(tagIdContentPOList);
            return tagIdContentPOPage;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    // 获取高亮的 SearchHit 对象中包含的标签内容
    private TagIdContentPO fromHighlightHit(SearchHit hit) {
        HighlightField highlightField = hit.getHighlightFields().get(CONTENT_FIELD_NAME);
        TagIdContentPO tagIdContentPO = new TagIdContentPO();
        tagIdContentPO.setTagId(Long.valueOf(hit.getId()));
        tagIdContentPO.setContent(highlightField == null ? "" : highlightField.getFragments()[0].string());
        return tagIdContentPO;
    }

    /**
     * 批量刷新 es 中的最热分数数据
     *
     * @param topScoreMap 最热分数数据映射，key：标签 id，value：标签的最热分数
     */
    public void refresh(Map<Long, Double> topScoreMap) {
        BulkRequest bulkRequest = new BulkRequest();
        topScoreMap.forEach((tagId, topScore) -> bulkRequest.add(new UpdateRequest()
                .index(INDEX)
                .id(String.valueOf(tagId))
                .doc(TOP_SCORE_FIELD_NAME, topScore)
        ));

        try {
            BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulkResponse.hasFailures()) {
                LOGGER.error(bulkResponse.buildFailureMessage());
            }
        } catch (IOException e) {
            // 解决无法解析 BulkResponse 的问题
            if (e.getMessage().endsWith("200 OK}")) {
                return;
            }
            throw new RuntimeException(e);
        }
    }
}
